package spring.cloud.tasks.client;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.ZooKeeper;
import spring.cloud.tasks.common.zookeeper.ZooKeeperDataManager;
import spring.cloud.tasks.common.zookeeper.ZooKeeperManager;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.Timer;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public abstract class Worker {

    @Setter
    @Getter
    public volatile String ip;

    @Setter
    @Getter
    public volatile String workerId;
    @Setter
    @Getter
    public volatile ZooKeeperManager zooKeeperManager;
    @Setter
    @Getter
    public volatile Lock lock = new ReentrantLock();


    @Setter
    @Getter
    public volatile boolean init = false;//zk初始化成功
    @Setter
    @Getter
    public volatile ZooKeepersConnectAndWorkerTasksInitThread zooKeepersConnectAndWorkerTasksInitThread;
    @Setter
    @Getter
    public volatile String initThreadErrorMessage;
    @Setter
    @Getter
    public volatile List<String> taskIdList = new ArrayList<String>();

    @Setter
    @Getter
    public volatile boolean executeTasks = true;

    @Setter
    @Getter
    public volatile Timer timer;
    @Setter
    @Getter
    public volatile WorkerTimerTask workerTimerTask;
    @Setter
    @Getter
    public volatile int timerInterval = 2000;
    @Setter
    @Getter
    public volatile String pongpongErrorMessage = "";


    public void init() throws Exception {
        Properties properties = new Properties();
        this.init(properties);
    }

    /**
     * 初始化
     */
    public void init(Properties properties) throws Exception {

        //
        this.lock.lock();
        try {
            if (this.zooKeepersConnectAndWorkerTasksInitThread != null) {
                this.zooKeepersConnectAndWorkerTasksInitThread.stopInitThread();
                this.zooKeepersConnectAndWorkerTasksInitThread = null;
            }
            if (timer != null) {
                timer.cancel();
                timer = null;
            }

            cleanZooKeeperDataManager();
            if (this.zooKeeperManager != null) {
                this.zooKeeperManager.close();
            }
            //
            Client.setWorker(this);
            this.zooKeeperManager = new ZooKeeperManager(properties);
            zooKeepersConnectAndWorkerTasksInitThread = new ZooKeepersConnectAndWorkerTasksInitThread(this);
            zooKeepersConnectAndWorkerTasksInitThread.setName("ItemTaskWorker-zooKeepersConnectAndWorkerTasksInitThread");
            zooKeepersConnectAndWorkerTasksInitThread.start();
        } finally {
            this.lock.unlock();
        }
    }


    /**
     * zk连接成功后进行初始化
     */
    public void initTasksAfterZooKeeperConnected() throws Exception {
        this.zooKeeperManager.doInit();
        initZooKeeperDataManager();
        List<String> taskIdList = this.getZooKeeperDataManager().registerWorkerToPermittedTaskList(this);
        log.info("注册成功taskIdList:" + JSON.toJSONString(taskIdList));
        //下面进行worker和zk交互逻辑
        if (this.timer != null) {
            this.timer.cancel();
            this.timer = null;
        }
        this.timer = new Timer("ItemTaskWorker-Timer");
        //
        if (this.workerTimerTask != null) {
            this.workerTimerTask = null;
        }
        this.workerTimerTask = new WorkerTimerTask(this);
        this.timer.schedule(this.workerTimerTask, 2000, this.timerInterval);
    }


    /**
     * zk连接不上才重启
     */
    public void reStart() throws Exception {
        try {
            this.executeTasks = false;
            if (this.zooKeepersConnectAndWorkerTasksInitThread != null) {
                this.zooKeepersConnectAndWorkerTasksInitThread.stopInitThread();
            }
            if (this.timer != null) {
                if (this.workerTimerTask != null) {
                    this.workerTimerTask.cancel();
                    this.workerTimerTask = null;//对象并行处理模型
                }
                this.timer.purge();
            }
            //
            stopAll();

            //
            if (this.getZooKeeperManager() != null) {
                this.getZooKeeperManager().close();
            }
            this.workerId = null;
            //
            this.init();
        } catch (Throwable e) {
            log.error("" + e.getMessage(), e);
        }
    }


    public void stop() throws Exception {
        try {
            lock.lock();
            this.executeTasks = false;
            if (this.zooKeepersConnectAndWorkerTasksInitThread != null) {
                this.zooKeepersConnectAndWorkerTasksInitThread.stopInitThread();
            }
            if (this.timer != null) {
                if (this.workerTimerTask != null) {
                    this.workerTimerTask.cancel();
                    this.workerTimerTask = null;
                }
                this.timer.cancel();
                this.timer = null;
            }

            stopAll();
            if (this.zooKeeperManager != null) {
                this.zooKeeperManager.close();
            }
            if (this.getZooKeeperDataManager() != null) {
                try {
                    ZooKeeper zooKeeper = this.getZooKeeperDataManager().getZooKeeper();
                    if (zooKeeper != null) {
                        zooKeeper.close();
                    }
                } catch (Exception e) {
                    log.error("stopAll zookeeper getZooKeeper", e);
                }
            }
            this.workerId = null;
            log.info("stopAll");
        } catch (Throwable e) {
            log.error("stopAll" + e.getMessage(), e);
        } finally {
            lock.unlock();
        }
    }

    public abstract void initZooKeeperDataManager() throws Exception;

    public abstract void cleanZooKeeperDataManager();

    public abstract ZooKeeperDataManager getZooKeeperDataManager();

    public abstract void stopAll() throws Exception;

    //
    public abstract void pongpong() throws Exception;


}
